package cn.my.springHDFS.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.SequenceFile;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static org.apache.hadoop.fs.permission.FsAction.ALL;

/**
 * Created by Leo on 2017/6/2.
 * HDFS API 操作
 */
@Component
@ConfigurationProperties(prefix = "hadoop")
public class HDFS_Util {
    //配置文件读取结果
    PropertyUtil property = new PropertyUtil();

    //默认地址
    private String HDFS_IP_ADDRESS = property.getProperty("master");

    private String HDFS_Url = "hdfs://" + HDFS_IP_ADDRESS + ":9000";

    /**
     * @author Leo by 2017.6.4
     * 新建文件夹
     * @param conf 配置文件
     * @return 返回结果
     */
    public boolean CreatDir(String dst ,Configuration conf, String username){
        HDFS_Url = "hdfs://" + conf.get("fs.default.name");
        try{
            //获取路径，以及配置
            FileSystem hdfs = FileSystem.get(new URI(HDFS_Url), conf, username);
            Path dstPath = new Path(dst);
            System.out.println("HDFS_Util: " + dstPath);

            //判断文件路径是否存在
            if(hdfs.exists(dstPath)){
                System.out.println("There is already exist!" + dstPath);
                return false;
            }

            //赋予权限
            FsPermission filePermission = new FsPermission(
                    ALL, //user action
                    ALL, //group action
                    FsAction.NONE); //other action

            //创建文件夹且带有权限
            hdfs.mkdirs(dstPath, filePermission);
            //设置文件夹所有者
            hdfs.setOwner(dstPath, username, null);
        }
        catch(IOException ie){
            ie.printStackTrace() ;
            return false ;
        } catch (InterruptedException | URISyntaxException e) {
            e.printStackTrace();
        }
        return true ;
    }


    /**
     * @author Leo by 2017.6.4
     * 文件上传
     * @param localFilesrc 本地文件路径
     * @param HDFS_dst HDFS路径
     * @param conf 配置文件
     * @return 返回结果
     */
    public boolean putToHDFS(String localFilesrc , String HDFS_dst , Configuration conf, String username){
        Path dstPath = new Path(HDFS_dst) ;
        try{
//            InputStream input = new BufferedInputStream(new FileInputStream(src));
            FileSystem hdfs = dstPath.getFileSystem(conf);
            hdfs.copyFromLocalFile(new Path(localFilesrc), dstPath);
            hdfs.setOwner(dstPath, username, "supergroup");
//            hdfs.setOwner();
//            FSDataOutputStream out = hdfs.create(new Path(dst), () -> System.out.println("."));
//            IOUtils.copyBytes(input, out, conf, true);
        }
        catch(IOException ie){
            ie.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     *  @author Leo by 2017.6.4
     * 文件下载(暂时没什么用)
     * HDFS的管理界面也是用的WebHDFS的下载接口
     * @param src 源地址
     * @param dst 目标地址
     * @param conf 配置文件
     * @return 返回结果
     */
    public boolean getFromHDFS(String src , String dst , Configuration conf){
        Path dstPath = new Path(dst) ;
        try{
            FileSystem hdfs = dstPath.getFileSystem(conf) ;
            hdfs.copyToLocalFile(false, new Path(src), dstPath) ;
        }catch(IOException ie){
            ie.printStackTrace() ;
            return false ;
        }
        return true ;
    }


    /**
     * @author Leo by 2017.6.4
     * 文件/文件夹删除
     * @param path 路径
     * @param conf 配置文件
     * @param username 用户名
     * @return 返回结果
     */
    public boolean checkAndDel(final String path , Configuration conf, String username){
        Path dstPath = new Path(path) ;
        try{
            FileSystem hdfs = dstPath.getFileSystem(conf);
            hdfs.setOwner(dstPath, username, "supergroup");
            if(hdfs.exists(dstPath)){
                hdfs.delete(dstPath, true) ;
            }else{
                return false ;
            }
        }catch(IOException ie ){
            ie.printStackTrace() ;
            return false ;
        }
        return true ;
    }


    /**
     * @author Leo by 2017.6.4
     * 获取HDFS所有DataNode的信息
     * @param conf 配置文件
     * @return 返回结果
     */
    public boolean getAllDataNodeName(Configuration conf){

        DistributedFileSystem hdfs;
        try {
            hdfs = (DistributedFileSystem) FileSystem.get(conf);
            DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();

            for (DatanodeInfo dataNode : dataNodeStats) {
                System.out.println(dataNode.getHostName() + "\t" + dataNode.getName());
            }

        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }

        return true;
    }


    /**
     * @author Leo by 2017.6.4
     * 给HDFS中的文件重新命名
     * @param oldName 文件旧名字(含路径)/文件夹路径
     * @param newName 文件新名字(含路经)/文件夹路径
     * @param conf 配置文件
     * @return Boolean
     */
    public boolean renameFile(String oldName, String newName, Configuration conf){
        try {
            FileSystem hdfs = FileSystem.get(conf);

            Path pathOldName = new Path(oldName);
            Path pathNewName = new Path(newName);

            return hdfs.rename(pathOldName, pathNewName);

        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * @author Leo by 2017.6.4
     * 查看某个文件在HDFS集群的位置
     * @param path 路径
     * @param conf 配置文件
     * @return 返回结果
     */
    public Map<String, String> WhereIsFileInCluster(String path, Configuration conf) throws IOException {

        FileSystem hdfs = FileSystem.get(conf);

        Path pathSrc = new Path(path);
        FileStatus fileStatus =  hdfs.getFileStatus(pathSrc);
        BlockLocation[] blockLocations = hdfs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        Map<String, String> blockMap = new HashMap<String, String>();
        for(BlockLocation block : blockLocations){
            blockMap.put(block.getHosts()[block.getHosts().length - 1], block.getNames()[block.getNames().length - 1].replace("50010", "50075"));
//            System.out.println(Arrays.toString(block.getHosts())+ "\t" + Arrays.toString(block.getNames()));
        }
//        for (Map.Entry<String, String> entry : blockMap.entrySet()) {
//            System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue().replace("50010", "50075"));
//        }
        return blockMap;
    }


//////////////////////////////////////////////// --- 以下虚线方法入口为 getFolderOrFileInPath --- ///////////////////////////////////////////////////////////////
    /**
     * @author Leo by 2017.6.4
     * 读取HDFS某个目录下的所有文件
     * @param path 路径
     * @param conf 配置文件
     * @return 返回JSON
     */
    public JSONObject getFolderOrFileInPath(String address, String path, Configuration conf) throws IOException, JSONException {

        FileSystem hdfs = FileSystem.get(conf);
        Path folderPath = new Path(path);
        FileStatus fileStatus[] = hdfs.listStatus(folderPath);

        List<JSONObject> fileItems = new ArrayList<JSONObject>();
        StringBuilder sb = new StringBuilder();

        JSONObject fileItem = null;
        for (FileStatus file : fileStatus) {
            fileItem = new JSONObject();
            String filePath = file.getPath().toString().replace(sb.append("hdfs://").append(address), "");

            if (file.isDirectory()){
                fileItem.put("text", file.getPath().getName());
                fileItem.put("path", filePath);
                _ValidateEmpty(address, conf, fileItems, file, fileItem, filePath);
            }
            else {
                fileItems.add(file_details(file, filePath));
            }
        }

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("json_data", fileItems);
        return jsonObject;
    }

    /**
     * @author Leo by 2017.6.10
     * 获得子节点
     * @param path 路径
     * @param conf 配置文件
     * @return 返回List JSON
     * @throws IOException IO错误
     * @throws JSONException JSON错误
     */
    public List<JSONObject> getChildNode(String address, String path, Configuration conf) throws IOException, JSONException {
        FileSystem hdfs = FileSystem.get(conf);
        Path folderPath = new Path(path);
        FileStatus fileStatus[] = hdfs.listStatus(folderPath);

        List<JSONObject> json_fileItems = new ArrayList<JSONObject>();
        StringBuilder sb = new StringBuilder();

        JSONObject fileItem = null;
        for (FileStatus file : fileStatus) {
            fileItem = new JSONObject();
            String filePath = file.getPath().toString().replace(sb.append("hdfs://").append(address), "");

            if (file.isDirectory()){
                fileItem.put("text", file.getPath().getName());
                fileItem.put("path", filePath);
                _ValidateEmpty(address, conf, json_fileItems, file, fileItem, filePath);
            }
            else if (!file.isDirectory()){
                json_fileItems.add(file_details(file, filePath));
            }
        }
        return json_fileItems;
    }

    /**
     * @author Leo by 2017.6.19
     * 封装文件信息
     * @param fileStatus 文件状态
     * @param path 路径
     * @return 返回JSON
     * @throws JSONException 抛出JSON错误
     */
    private JSONObject file_details(FileStatus fileStatus, String path) throws JSONException {
        SimpleDateFormat dt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        JSONObject file_details_item = new JSONObject();
        file_details_item.put("text", fileStatus.getPath().getName());
        file_details_item.put("path", path);
        file_details_item.put("rights", fileStatus.getPermission().toString());
        file_details_item.put("date", dt.format(new Date(fileStatus.getModificationTime())));
        file_details_item.put("size", fileStatus.getLen());
        file_details_item.put("type", "file");
        file_details_item.put("level", appearNumber(path, "/"));
        return file_details_item;
    }

    /**
     * @author Leo by 2017.6.11
     * 验证是否为空文件夹
     * @param address 地址
     * @param conf 配置文件
     * @param fileItems 多个文件项目
     * @param file 文件
     * @param fileItem 文件项目
     * @param filePath 文件路径
     * @throws IOException
     * @throws JSONException
     */
    private void _ValidateEmpty(String address, Configuration conf, List<JSONObject> fileItems, FileStatus file, JSONObject fileItem, String filePath) throws IOException, JSONException {
        List<String> tagsNumber = new ArrayList<String>();
        tagsNumber.add(String.valueOf(checkIsEmptyFolder(filePath, conf)));
        fileItem.put("tags", tagsNumber);
        if (Integer.valueOf(tagsNumber.get(0)) > 0){
            fileItem.put("nodes", getChildNode(address, file.getPath().toString(), conf));
            fileItem.put("IsEmpty", false);
            fileItem.put("type", "dir");
            fileItems.add(fileItem);
        }
        else {
            fileItem.put("IsEmpty", true);
            fileItems.add(fileItem);
        }
    }

    /**
     * @author Leo by 2017.6.10
     * 判断文件夹是否为空
     * @param path
     * @param conf
     * @return
     * @throws IOException
     */
    public int checkIsEmptyFolder(String path, Configuration conf) throws IOException {
        FileSystem hdfs = FileSystem.get(conf);
        Path folderPath = new Path(path);
        FileStatus fileStatus[] = hdfs.listStatus(folderPath);
        return fileStatus.length;
    }

    /**
     * @author Leo by 2017.6.11
     * HDFS API之外的方法
     * 获取指定字符串出现的次数
     *
     * @param srcText 源字符串
     * @param findText 要查找的字符串
     * @return
     */
    public int appearNumber(String srcText, String findText) {
        int count = 0;
        Pattern p = Pattern.compile(findText);
        Matcher m = p.matcher(srcText);
        while (m.find()) {
            count++;
        }
        return count;
    }

//////////////////////////////////////////////// --- 代码块止 --- ///////////////////////////////////////////////////////////////

}
